package com.zyt.hbase.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.NavigableMap;

/**
 * @author zyt
 * @createDate 2020/09/15 9:20 AM
 */
public class MyProcessor extends BaseRegionObserver {
    @Override
    public void preDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        familyCellMap.forEach((byte[] key, List<Cell> value) -> {
            value.forEach(item -> {
                try {
                    int qualifierOffset = item.getQualifierOffset();
                    int qualifierLength = item.getQualifierLength();
                    byte[] qualifierArray = item.getQualifierArray();
                    byte[] qualifierArrs = new byte[qualifierLength];
                    System.arraycopy(qualifierArray, qualifierOffset, qualifierArrs, 0, qualifierLength);
                    String c = new String(qualifierArrs);
                    System.out.println("列名：" + c);

                    //删除对应数据
                    final HTable friendsTable = (HTable) e.getEnvironment().getTable(TableName.valueOf("relationship"));
                    final Delete delete1 = new Delete(Bytes.toBytes(c));
                    delete1.addColumn(Bytes.toBytes("friends"), delete.getRow());
                    //执行删除
                    friendsTable.delete(delete1);
                    //关闭table对象
                    friendsTable.close();
                    System.out.println("删除关联数据！！" + c + ":" + new String(delete.getRow()));
                } catch (Exception exception) {
                    exception.printStackTrace();
                }

            });
        });
    }
}
